流式超越批量：Streaming 102翻译

译者注

在流式系统方面，网络上流传了两篇精彩的文章，Streaming 101 与 Streaming102。这两篇文章的作者是Google流式系统的负责人Tyler Akidau，他是MillWheel与DataFlow的开发者，在流式系统方面十分权威。恰巧这两篇文章非常详细并且适合初学者，可以帮助我们理清流式系统的各种概念与面临的挑战，是十分难得的佳作。其中Streaming101已经有很多翻译的版本，质量参差不齐。官方的翻译版本参考这里，然而不幸的是它依然晦涩难懂。这里并不打算翻译101这篇文章，读者可以参考原文和译文进行学习，它主要介绍了流式系统的一些基本概念和困境，同时提出了一个核心观点：设计优良的流式系统完全可以代替批量系统，并从原理上分析了如何实现这一目标。在流式102这篇文章中，作者以实际的流式系统（DataFlow）的设计为例，列举了多个场景，具体讲解了如何设计流式系统，解决这些问题。这里将翻译流式102这篇文章，我会参考原文采用意译的方式，力求更好的帮助读者理解。

在阅读流式102之前，请务必通篇阅读流式101，理解核心概念，这里的思维导图仅供参考。Let’s Go!

简介

欢迎回来！如果您错过了之前的《The world beyond batch: Streaming 101》，强烈推荐先阅读这篇文章，它是理解下文的基础。请注意：我们假设你已经熟悉了那些技术术语和概念。此外，本文有一些动画Gif，如果你用打印出来看就会丢失这些信息。

让我们进入正题。简要回顾一下，上一次我们主要讨论了3个主题：

技术术语：精确定义了一些已经已经被大家说烂了的词，比如『流式』（Streaming）
批量VS流式：比较了批量系统与流式系统的能力。并假定只要做好了准确性(Correctness)和时序推理工具(Tools For Reasoning About Time)这两件事，流式系统就可以超过批量系统。
数据处理的模式/套路：分别介绍了用流式系统(Streaming)和批量系统(Batch)来处理有界(Bounded)和无界(Unbounded)数据集的常用方法。

这篇文章中，我们会接着深入探讨数据处理模式，但是会更加关注细节和具体的应用场景。本文主要分为两个章节：

流式101的精简版：简要回顾Streaming 101中引入的概念，并引入一个Demo来突出一下重点。
流式102：流式101的姊妹篇，详细分析处理无界数据集的一些其他相关概念。同样会用一些Demo来解释概念。

当此文完结，我们就会掌握构建一个健壮的处理无序数据的系统所需的核心原理与概念。有一些时序推理工具可以让你真正地超越批量系统！😏

为了给你提供一些直观感受，我会使用DataFlow SDK的代码片段（例如：Google DataFlow API）。同时，我还用了一些动画来可视化一些概念。至于为何使用DataFlow而不是Spark Streaming或者Storm这类系统，其一是我更熟悉DataFlow，其二是其他系统并没有这种表现能力来展示我所涉及的例子（潜台词：功能不全面）。好消息是其他系统也在朝着这个方向努力，一个更大的好消息是Google现在已经向Apache基金会提出Propsal来建立一个DataFlow的孵化器项目（与Artisans, Cloudera, Talend等公司合作）。它的目标旨在以由DataFlow模型提供的强大的无序数据处理语意为基础来建立一个开源社区和生态。这会在2016年渐渐变得有趣，我跑题了。。。

抱歉，本文中没有上次我承诺的各个流式框架比较那部分。我错误估计了本文包含的内容以及时间，我不能再拖稿了。作为补偿，最后会附上我在2015年新加坡Hadoop大会上发表演讲《The evolution of massive-scale data processing》（在2016的六月会更新版本），它有一些材料就是我想讲的。同时，奉上一个精美的幻灯片，Enjoy。

现在，开始讨论流式！

回顾&&大纲

在流式101中，我首先澄清了一些技术术语：

bounded数据 VS unbounded数据。有界数据源通常是有限的大小，经常指的是『批量』数据。无界数据源通常有无限的大小，通常指的是『流式』数据。我会尽量避免使用『批量』或者『流式』这些词来描述数据源，他们的含义会引起无解和导致不必要的限制。
然后，我进一步定义了流式引擎与批量引擎的不同点。批量引擎是仅仅为处理有界数据而设计的，而流式引擎则为无界数据而设计。我只会用『流式』和『批量』这些词来表示执行引擎。

紧接着我们又涉及了与无界数据处理相关的两个重要概念：

首先我们要严格区分事件时间(Event Time，事件真实发生的时间)和处理时间(Processing Time，事件被我们观察到的时间)。在流式101一文中我们认为这是重要的基础之一：如果您关心正确性和事件上下文，则必须使用事件时间，而不是处理时间。
我们又介绍了窗口化(Windowing，沿着时间边界划分数据集)的概念。这是一种解决无限数据源永远不会结束这一问题的常见方法。窗口化策略的简单的方法有固定窗口和滑动窗口，但还有一些更复杂的窗口类型，如会话（该窗口由数据本身的特征定义，例如，使用每个用户不活动的间隙来切分）也有广泛的应用。

除了101中介绍的两个重要概念，本文我们还将引入另外三个概念

Watermarks：Watermark(水印)是用来表示与事件时间相关联的输入完整性的概念。对于事件时间为X的Watermark是指：已经观察到事件时间小于X的所有输入数据。因此，当观测对象是没有尽头的无界数据源时，Watermark来测量数据进度。
Triggers：触发器是一直外部信号触发机制，用于表示什么样的信号会真正地触发窗口中的数据被计算。触发器为数据计算的触发提供灵活性。触发器还使我们可以在有需要时多次运算数据（例如数据在某时间更新了，要重算）。这为我们可以随着时间变化不断优化运算结果打开了方便之门：我们可以在数据到达时先提供一个推测(speculative)版本的结果，然后，随时间推移不断处理上游数据的变换（生成了修订版本）或者处理那种在Watermark之后才到达的数据（例如，手机的场景：某人在断网时记录各种动作及其事件时间，然后在重新联网后，上传这些事件进行处理。）
Accumulation：累加/更新模式指定在同一窗口中观察到的多个运算结果之间的关系。这些结果之间可能完全不相关，例如与时间先后无关的结果，直接覆盖以前的运算结果即可。不同的累积模式具有不同的语义和成本，视各种场景而定。

最后，为了更好地理解所有这些概念之间的关系，我们将重新回顾全文，并通过回答下面四个问题来温故知新。我提出的这几个问题都至关重要：

What results are calculated?：做什么计算？（计算逻辑是什么，如何表示？）这个问题的答案是：pipeline中的各种transform操作。这包括求和，建立直方图，训练机器学习模型等等。这也是经典批处理回答的问题。
Where in event time are results calculated?：计算什么时间范围的数据？（这里的时间指的是Event-Time）这个问题的答案是：在pipeline中用EventTime来窗口化数据。这包括流式101中提到的窗口化方法（固定窗口，滑动窗口，会话窗口），以及似乎没有窗口概念的场景（例如，Streaming 101中描述的时间不可知处理;经典批处理也通常属于此类别。此外，还有其他更为复杂的窗口类型，如时间有限的拍卖（time-limited auctions）。请注意，如果在进入系统的时间标记为事件时间，上面这些清空也可以指的是Processing Time。
When in processing time are results materialized? ：何时将计算结果输出？这个问题的答案是：使用watermark和trigger配合触发计算。这有很多的组合，但最常用的方法（最佳实践）是用Watermark表示某个窗口的输入完成，同时，配合允许Eearly Data（用于在窗口完成之前发出的推测结果）和Later Data（水印只是完整性的估计，当水印指示窗口的输入完成之后，更多的输入数据还是可能会到达）的触发器。
How do refinements of results relate?：后续数据的处理结果如何影响之前的处理结果？这个问题通过Accumulation来解决：丢弃（结果之间是独立且不同的），累积（后来的结果建立在先前的结果上）或累积并撤回（其中累积值加上先前触发的值的撤回）

下文会详细讨论这些问题：我会用上一些配色以清晰地表示我们讨论的问题属于上面的哪一种 What/Where/When/How。

流式101的精简版

首先，我们来回顾一下Streaming 101中提出的一些概念，但这次我将用一些详细的例子使得这些概念更具体。

What: transformations

经典批处理应用的转换操作已经回答了第一个问题：“计算逻辑是什么？”。尽管许多人可能已经熟悉经典的批处理，但我们将从那里开始，因为它是我们的基础，我将基于他添加一些其他概念。

这一节我们就看一个例子：在一个简单的数据集（如10个值）上按key聚合求SUM。你可以把它想象成某个手游里面通过把个人的分数求和来计算团队总分，或者其他计费类应用或者Monitor的应用等。

对于后面的每一个例子我都会用Dataflow的伪代码来定义具体的Pipeline。这将是伪代码，在某种意义上，我有时会稍稍改变规则，使示例更清晰，去除不必要细节（如使用的具体I / O源）或简化命名（Java中的当前触发器名称是否繁琐，为了清晰起见，我将使用更简单的名称）。除了这些小事情之外（大部分我在后文注释中明确列举），它基本上是真实可用的Dataflow SDK代码。稍后对那些对他们可以编译和运行自己的类似示例感兴趣的人，我还将提供一个链接到实际的代码。

如果你熟悉Spark Streaming或者Flink，那么比较容易理解DataFlow的代码，下面简单介绍DataFlow的两个基本原语：

PCollections：表示可以执行并行（这里的P表示的含义）的转换（Transform）操作的数据集（可能非常大）
PTransforms：应用到PCollection上，来执行的转换操作，生成新的PCollection。PTransforms可以是对元素一个一个操作，也可以是聚集（agg）操作，或者可以与其他的PCollection相互组合。

如果有疑问或者想查看DataFlow的文档，看这里。

为了讲解例子的方便，我们这里使用了一个名叫input的PCollection<KV<String, Integer>>作为输入（input是由String/Integer作为键/值对组成的，String是球队名，Interger是每人的分数）。在实际的Pipeline中，input一般从I/O源读入原始数据，然后解析日志数据成Key/Value对，最终转换成PCollection<KV<String, Integer>>。我会在第一个示例中写出所有代码，但是在接下来的其他示例中隐去I/O相关的部分。

因此，一个例子：要求先从数据源中解析出team/score键值对，然后对每个team求和，算出球队总分。代码如下：

// Listing 1. Summation pipeline
PCollection<String> raw = IO.read(...);
PCollection<KV<String, Integer>> input = raw.apply(ParDo.of(new ParseFn());
PCollection<KV<String, Integer>> scores = input
  .apply(Sum.integersPerKey());

对于下面的所有例子，我们都会先分析代码，然后用动画展示一个数据集的运行过程。具体点说，就是含有一个key和10条数据的pipeline的执行过程。在实际运行时，会有多台机器并行执行相同的操作。这里仅仅是为了简单清晰地讲解例子。

所有动画都会在两个维度上绘制输入和输出：事件时间（X轴）和处理时间（Y轴）。因此，从pipeline的视角来看，白色的粗线条从底部向上移动代表了真实时间的移动。输入用圆圈表示，圆圈内的数字表示该记录的值。他们一开始是灰色的，pipeline实际观测到他们后会改变颜色。

随着pipeline观测到这些值，会逐渐在状态（State）中累加他们，并最终实际输出结果。状态的变换和输出都由矩形表示，聚合值在矩形上面，被矩形覆盖的区域表示该部分的事件时间/处理时间已经累积计算到了结果中。在批处理引擎中执行Listing 1中的pipeline代码，运行过程如下（请注意，您需要点击/点击下面的图像才能启动动画，然后再循环直到再次点击/再点击）：

图2：传统的批量处理过程

由于这是一个批量的pipeline，因此，只有接收到所有的input值，系统才会累加状态（由图中顶部的绿色虚线表示），最终产生了唯一的输出：51。而且，这个例子由于没有使用任何Window，我们计算了所有Event-Time内的值的和。所以，图中用于表示状态和输出的矩形覆盖了整个X轴。但是，如果我们要处理一个无界数据源，那么传统的批量处理就不行了；我们不能等待输入数据结束，因为它实际上永远不会结束。因此，我们需要引入一个在Streaming101中提及的概念：窗口。在回答第二个问题，“计算什么时间范围的数据？”之前，我们先简要回顾一下窗口。

Where: windowing

如上次讨论的那样，窗口化是沿着时间边界分割数据源的过程。常见的窗口划分策略包括固定窗口，滑动窗口和会话窗口。

看一个实际的例子：把上面的求和的pipeline划分为2分钟的固定时间窗口。使用DataflowSDK，添加一个 Window.into transform操作即可：

// Listing 2. Windowed summation code.
PCollection<KV<String, Integer>> scores = input
  .apply(Window.into(FixedWindows.of(Duration.standardMinutes(2)))) // 这一行
  .apply(Sum.integersPerKey());

回想一下，由于语义上批处理只是流式的一个子集，所有Dataflow提供了一种统一的批处理和流式模型。因此，我们首先在批量引擎上执行这个pipeline，这种机制看起来更直观。当我们切换为流式引擎时，可以直接与这个进行比较。

图4：在批量引擎上对窗口进行求和

如前所述，直到输入数据被完全读入，状态才会累加，之后产生输出。但是，在这种情况下，我们得到的不是一个输出值，而是四个输出值：每个输出对应着2分钟的事件时间窗口。

至此，我们回顾一下Streaming 101中介绍的两个主要概念：事件时间（Event-Time）与处理时间（Processing-Time）的关系以及窗口化（Windowing）。如果我们想深入学习，我们需要增加本节开头提到的新概念：水印（Watermark），触发器（Trigger）和累积（Accumulation）。因此开始Streaming 102。

流式102

我们刚刚看到了如何在批量引擎上运行窗口化的pipeline。但是，我们更希望系统有更低的延迟并且可以原生支持处理无界数据源。切换到流式引擎是正确的方向！但问题是，在批量引擎上我们清楚地知道什么时间点数据完整了（比如：当有限数据源的数据全都读入的时候），但是对于无界的数据源，我们缺乏一种有效的方式来判断数据完整性（Completeness）。因此，引入了Watermark

When: watermarks

Watermark可以给出”何时将计算结果输出？”这个问题的一半答案：Watermark是在Event-Time域上的时间概念，用来刻画输入完整性。（这句话说明，watermark首先是表示的是一个Event-Time时间）。换句话说，它们是系统以Event-Time为尺度来衡量事件流中Record处理进度/完整性的方式（不管是无界数据还是有界数据都适用，显然在无界的情况下更有用）。

回想一下Streaming 101中的这个图，这里稍作修改，其中描述了事件时间和处理时间之间的偏差（skew），在真实的分布式系统中，这个偏差会随时间不断变化。

上面这个红色曲线就是真实的Watermark，随着Processing-Time的推移，他描述了Event-Time纬度的完整性的过程。你可以把Watermark看成是 F(P) -> E的函数：输入是Processing-Time，输出是Event-Time。（确切的说，函数的输入是在pipeline中被观测到的Watermark这一点的所有上游的当前状态：输入源，缓冲数据，正在处理的数据等；但在概念上，将其视为从Processing-Time到Event-Time的映射更为简单。）在Event-Time上的这一点E表示：系统相信在E之前的所有数据都被观测到了。换句话说，系统『确信』不会再有Event-Time<E的数据出现了。根据这种『确信』是不是严格保证或者仅仅是猜想，我们把Watermark分为两种类型：完美Watermark与启发式Watermark。

完美Watermark：如果我们对所有输入数据十分了解，就有可能构建一个完美的Watermark。这种情况下，输入源不存在数据迟来的问题，所有数据只会提前或者准时到达。
启发式Watermark：对于大部分的分布式输入源，期望对输入数据完全掌控是不切实际。这种情况下一般使用启发式的Watermark。启发式Watermark会使用一切可用的信息（包括分区，分区内排序，文件增长速度等）来尽量准确地推断输入的进度。在许多情况下，这种Watermark也可以预测了非常准确。即使如此，启发式Watermark的意味着它的预测有时可能是错误的，这将导致迟到的数据。我们将在下面的Trigger部分中了解如何处理迟来的数据。

Watermark是一个有意思而且复杂的话题，超出我的讨论范围，期望未来有时间写个帖子讨论它。现在，为了更好地了解Watermark的作用和缺陷，我们使用Watermark的流式引擎，来看看Listing 2中的pipeline代码何时将计算结果输出。左边的例子使用完美的Watermark，右边的使用了启发式Watermark。

图6：在流式引擎上分别使用完美watermark（左）与启发式watermark（右）进行分割窗口求和的运算

这两种情况都是watermark到达窗口终点时结果被输出（注：看图中watermark的先与12:02 12:04 12:06 12:08的交点）。主要的不同在于：右侧启发式算法没有计算『9』这个值，这对watermark曲线的形态影响巨大。这个例子突出了watermark的两个缺点（不仅watermark，任何一种完整性概念都有类似的问题），具体来说：

太慢了：当watermark因为知道有数据还没到来被正当地推迟了（比如：网速下降导致输入变慢），如果我们最终输出结果仅仅依赖于watermark曲线的提升，那么这种延迟会直接影响了系统最终输出的时间。这在左图中体现的最明显：即使后面窗口的数据都已经到来了，第一个窗口迟到的数据『9』卡住了后续所有窗口的watermark线。第二个窗口[12：02，12：04)的问题尤其突出，从窗口中的第一个值到达，直到输出结果，需要将近七分钟。在右图中的启发式watermark中，这个问题不是很严重（延迟5分钟分钟），但是不意味着启发式算法没有这个问题；这仅仅是因为这个例子中我选择启发式watermark比较特殊。这里我们有一个重要的结论：虽然watermark提供了完整性的概念，但是他对系统延迟影响很大。假设你有一个显示数据指标的面板，你一定不能忍受要等待一个小时或者一天才能看到当前窗口的数据出现—这是批量系统才会有的痛点嘛。我们更希望可以随着时间推移看到指标逐渐变化，最终产生完整地结果。
太快了：启发式的watermark算法不能正确地标识数据。某些Event-Time小于watermark的数据会迟到地到达，产生了late data。这就是右图中发生的问题：在第一个窗口中，watermark线在所有数据到来之前就提前穿过了窗口终点，最终导致了错误的计算结果『5』（正确应该是14）。这是启发式算法导致的严重问题。『启发式』天然地隐含了这个意思—我有时会出错。所以，如果我们关心结果的正确性，仅仅使用启发式watermark是不够。

在Streaming 101中，我强调过：仅仅想使用完整性（Completeness）这个概念来健壮地处理无序无界的数据流还远远不够。watermark太慢或太快这两大缺点就是我这个观点的依据。仅仅依靠完整性的系统是不能同时获得低延迟和正确性的，解决这些问题的关键是引入触发器（Trigger）。

When: The wonderful thing about triggers, is triggers are wonderful things!（触发器是个好东西）

“何时将计算结果输出？”这个问题的另一半答案就是触发器。触发器声明了一个窗口的计算结果什么Processing-Time时间被输出？（但是，触发器自身做出决定可能依据的是其他Event-Time时间域发生了什么，比如watermark线的进度）。窗口内的每次特定输出被称为窗口的窗格（pane）。

触发Trigger的信号包括下面这些：

Watermark的进度（如：Event-Time的值）：其实这就是图6隐含的触发器，当watermark线到达窗口终点时触发输出。另外一个例子是：当触发器存在超过了一定时间后进行触发器的垃圾回收，后面我们会看到他的应用。
Processing-Time 的进度：用来提供定期更新数据，因为Processing-Time（不像Event-Time）总是大致均匀地移动，而不会出现延迟。
到达元素的数量：窗口中观察到一些有限数量的元素之后进行触发
特殊的标记：在Record的一些记录或特征值（例如，EOF元素或刷新事件）指示应该生成输出。

除了使用某个特定信号的简单触发器之外，还有组合触发器，允许创建更复杂的触发逻辑。组合触发器包括：

重复触发（Repetitions）：特别适用于Processing-Time触发器以提供定时更新操作。
联合触发（Conjunctions） ：(逻辑 AND)，只有所有子Trigger触发才会触发（例如，在watermark通过窗口终点并且我们观察到含有EOF符号的Record）
各自触发（Disjunctions）： (逻辑 OR)，只要有一个子触发器触发，就会触发 (例如，只要在watermark通过窗口终点或者我们观察到含有EOF符号的Record就触发).
按顺序触发（Sequences）：以预定义的顺序触发子触发器。（后一个子触发器必须等待前一个触发器触发）

为了让Trigger这个概念更具体，让我们明确地表示出图6中（也是Listing2的代码中）隐含的触发器：

// Listing 3. Explicit default trigger.
PCollection<KV<String, Integer>> scores = input
  .apply(Window.into(FixedWindows.of(Duration.standardMinutes(2)))
               .triggering(AtWatermark())) // 这一行
  .apply(Sum.integersPerKey());

考虑到这一点，基于对触发器功能的基本了解，我们可以思考来解决watermark太慢或太快的问题。对于这两个问题，我们基本上希望对于一个窗口可以定期更新输出值（即除了水印线到达窗口终点之外的更新）。因此，我们需要一些重复触发（Repetitions）器。那么这个问题就变成了：重复触发的条件是什么？

对于输出太慢的情况（即希望提供早期的推测结果的情况），我们可以假设给定窗口的早期阶段，流入数据量是稳定且不完整的。因此按照Processing-Time周期性触发（例如1分钟一次）是明知。因为触发的次数与数据量无关，最坏的情况就是获得稳定不变的触发数据流。

对于输出太快的情况（即对于启发式watermark可以处理迟到的数据），我们可以假定watermark是基于比较准确的启发式算法（这个加上还是比较靠谱的）。在这种前提下，我们不会经常看到late data到达，但是如果出现了这种情况，我们必须快速处理更新输出的结果（例如，看到late data，立刻更新）。因为我们假设这种情况很少见，所有他不会使得系统过载。

注意，这里只是举了一个例子，你可以根据自己的情况自由选择触发的条件（比如只在上面某一种情况下触发，或者都不触发）

最后，我们需要编排好各种触发器的时间：比watermark早触发，在watermark达到时触发还是比watermark晚触发？我们可以通过一个Sequence触发器和一个特殊的OrFinally触发器来实现，OrFinally触发器可以安装一个子Trigger，当子Trigger触发时终止父Trigger。

// Listing 4. Manually specified early and late firings.
PCollection<KV<String, Integer>> scores = input
  .apply(Window.into(FixedWindows.of(Duration.standardMinutes(2)))
               .triggering(Sequence( // 这两行
                 Repeat(AtPeriod(Duration.standardMinutes(1)).OrFinally(AtWatermark()), 
                 Repeat(AtCount(1))))
  .apply(Sum.integersPerKey());

但是，这样写起来有些麻烦。由于这种repeated-early | on-time | repeated-late的模式十分常见，我们在Dataflow API中提供了特定的简写方式（语意上等价的）。

// Listing 5. Early and late firings via the early/late API.
PCollection<KV<String, Integer>> scores = input
  .apply(Window.into(FixedWindows.of(Duration.standardMinutes(2)))
               .triggering(
                 AtWatermark()
                   .withEarlyFirings(AtPeriod(Duration.standardMinutes(1))) // 这两行
                   .withLateFirings(AtCount(1))))
  .apply(Sum.integersPerKey());

在流式引擎上执行Listing4/5的代码（使用完美watermark和启发式watermark）产生的结果如下：

图7.在安装了early/late触发器的流式引擎上进行窗口求和运算

这个版本比图6的版本有两处明显的提升：

对于第二个窗口[12: 02，12:04 ]中的“ watermark导致输出太慢 ”的情况：我们提供了在early阶段每分钟一次的定期更新。在完美watermark的案例中，差异最为明显，其中首次输出时间从近七分钟降至三分半；在启发式案例中也有明显改善。这两个版本现在随时间移动会提供稳定的不断改进的结果（窗格pane提供了7，14，22这些值的输出），同时我们做到了在输入数据完整和最终结果输出之间的低延迟。
对于第一个窗口[12:00，12:02]中的“ 启发式watermark太快输出 ”的情况：当『9』这个值晚点到达时，我们立即更新了输出结果，输出了正确的窗格值『14』。

一个意外的惊喜是：新加入的触发器机制让两种类型的watermark算法产生了统一的输出。对比图6的两个算法版本还截然不同，在这里的他们却十分相似。

这两个版本还有一个很大的区别：他们窗口的生命周期不同。完美watermark的案例中，我们知道watermark通过窗口终点后，系统不会再看到该窗口的任何数据了，因此我们可以关闭当前窗口的所有状态。在启发式watermark案例中，我们仍然需要让窗口等待一段时间来处理late data。但是问题是：到目前为止，我们没有办法知道需要保留窗口多长时间。这里我们需要引入『允许迟到时间』这个概念（也叫horizon）。

When: allowed lateness （垃圾回收，何时关闭Window）

在进入最后一个问题”后续数据的处理结果如何影响之前的处理结果？”之前，我们要讨论一下处理长期无序数据数据流的流式系统必备的一个功能：垃圾回收。图7的启发式算法watermark的例子中，每个窗口的状态在该示例的整个生命周期内都会保持。为了处理late data，这么干是必要的。但是，在实际环境中，当处理无限数据源时，无限期地保持窗口状态（包括元数据）是不切实际的，我们最终会耗尽磁盘空间。

因此，任何实际的无序处理系统都需要提供一些限制窗口生命周期的方法。一个简单的办法是在系统内定义一个允许数据迟到的视界（horizon，理解成时间范围） — 例如：对记录数据可以影响处理流程的时间（相对于watermark）进行限制；任何在这个时间点后到达的数据都会被简单地抛弃。一旦你划定了允许数据可能有多晚到达，你就准确地确定了窗口状态需要保持的最长时间，这段时间就是watermark线到达窗口终点线之后再继续等待的时间。此外还给予系统尽快丢弃超过horizon的数据的自由，这意味着不要在我们不关心的数据上浪费任何资源。

译者注：什么时间关闭窗口（垃圾回收）？
watermark超过窗口结束的延迟时间（EventTime_diff）。（我们定义）即传入当前ProcessTime，算出EventTime2，我们指定一个EventTime_diff，对于EventTime1的窗口，如果EventTime2-EventTime1>=EventTime_diff,就可以关闭之前的窗口了。

为什么用watermark（Event-Time）来做呢？而不是Processing-Time（比如在watermark到了之后等待n秒钟的Processing-Time）？
参考文章给出的注释：因为可能有其他原因导致系统崩溃延迟等等，使得processTime就这么过去了，窗口过早地关闭！

由于『允许迟到时间』和『watermark』之间的相互作用有点微妙，所以值得再举一个例子。我们来看一下Lising 5 /图7中的启发式watermark的pipeline的例子，我们添加一分钟的允许迟到时间（视界，horizon）（请注意，这个特定的horizon被选择是因为适合图标展示；在实际工程中，更大的horizon可能会更好一些）：

// Listing 6. Early and late firings with allowed lateness.
PCollection<KV<String, Integer>> scores = input
  .apply(Window.into(FixedWindows.of(Duration.standardMinutes(2)))
               .triggering(
                 AtWatermark()
                   .withEarlyFirings(AtPeriod(Duration.standardMinutes(1)))
                   .withLateFirings(AtCount(1)))
               .withAllowedLateness(Duration.standardMinutes(1))) // 这一行
  .apply(Sum.integersPerKey());

上面这个pipeline的执行如下面的图8所示，我给图标添加了下面这些特性来强调『允许延迟时间』的影响：

粗白线的位置指示当前Processing-Time，又给所有的窗口添加了有注释的虚白线来表示允许延迟的视界时间（注意延迟用Event-Time定义的）
一旦一个窗口内的时间通过watermark后再经过horizon，窗口就会关闭，意味着窗口的状态就好被抛弃。我会留下一个虚线框表示当窗口关闭时，它覆盖的时间范围（两个时间域都有，右侧多出的虚线小尾巴与watermark相交）。
对于这个图，我额外地为第一个窗口添加了一个迟到值『6』。『6』是迟到的，但仍然在允许延迟时间（horizon）范围内，所以它被并入更新的值为11的结果。然而，迟到的『9』超过了延迟的地平线，所以它被简单地抛弃了。

图8. 运行在流式引擎上，具有早期和晚期启动并且允许1分钟延迟的窗口求和

关于『允许延迟时间』（horizon）的最后两点说明：

如果使用完美的watermark，就不需要处理late data，这只允许延迟时间为0秒就行，这就是图7中所示的效果
某些特点的任务：如一些根据某些key计算agg的需求（如统计不同浏览器的访问量）。只要key的个数不是太多，没必要设置horizon。系统只是保持key对应的值的数量的window是活动的就行。
（译者注：推测这种情况下系统不是按照时间来分割window的。）

好了，让我们进入第四个也是最后一个问题。

How: accumulation

随着时间的推移，触发器会在一个Window中产生多个Pane。我们遇到了最后一个问题：『后续数据的处理结果如何影响之前的处理结果？』。迄今为止的例子中，每个窗格的新数据都建立在紧邻的前一个窗格的数据之上。但是，实际上有三种不同的更新模式。（注释：事实上有四种模式，第四种是丢弃并更正—discarding and retracting，这并不常用，这里不会讨论它）

丢弃（Discarding）：每当有窗格输出，过去的状态就会被丢弃，这意味着后续的窗格与之前的无关。应用场景是：下游消费者会自己进行累计更新操作。例如：那些只希望接收到差值（delta）来求和的系统，我们只需要把接收到的新数值发送给他就行。
累计（Accumulating）：如在图7中，每一个窗格输出时，过去状态被保留，和未来的输入一起累加形成新的当前状态。这意味着每个后续的窗格建立在前面的窗格之上。累加模式应用在这种场景：后续结果可以简单地覆盖以前的结果，例如把输出的结果存储在BigTable或HBase这类key/value存储中。
累计并更正（Accumulating & retracting）:与累计模式类似，但是当产生新的窗格时，它会再单独产生一个被更正/回撤的值。(新的累加值Y,更正值X)的组合相当于在说:”之前告诉你的结果X是错的，别管那个值了，用新的值Y代替它吧！”下面这两种情况更正值特别有用：
- 当下游消费者需要把不同维度的数据重新分组时，新产生的值完全有可能会与旧值不同，因此新数据最终会在不同的组中。在这种情况下，新值不能覆盖旧值; 您需要从旧组中删除旧值，而在新组中加入新产生的值。
- 使用动态窗口时（例如Session窗口，下面我们会研究它），由于可能发生窗口合并，新产生的窗口会替换多个旧窗口。我们无法从新窗口推知要替代哪些旧窗口，因此最好明确告知我们要被撤回的窗口有哪些。
让我们并排对比这些情况就更加明确了，查看图7中第二个窗口的3个窗格，下表展示了三种累加模式中每个窗格的值将如何变换。

| 放弃 | 积累 | 累积与收缩 | |
| ——————- | ——— | ————- | ——— |
| 第1页：[7] | 7 | 7 | 7 |
| 窗格2：[3,4] | 7 | 14 | 14，-7 |
| 窗格3：[8] | 8 | 22 | 22，-14 |
| 最后观察值 | 8 | 22 | 22 |
| 总和 | 22 | 51 | 22 |
丢弃：每个窗格仅包含当前到达的值。因此，最终观测值不是sum。但是，如果要自己单独计算所有窗格，就会就会得到正确的答案22。这就是为什么丢弃模式在下游消费者自己会执行某种聚集操作时很有用。
累计：如在图7中，每个窗格与之前已经到达的值相结合，加上从先前的窗格中的所有值。因此，最终观测值就是22。但是，如果你自己又单独地求了一次和，那么会给出不正确的51。这就是为什么当你可以用新值简单地覆盖以前的值时，累积模式是最有用的：新的值已经包含了迄今为止所有的数据。（译者注：类似reduce函数）
累计和缩回：每个窗格都包含一个新的累积模式值以及前一个窗格值的撤回值。因此最后最终观测值（不含撤回值）值以及最终所有窗格观测值（包括撤回值）的总和都能提供正确的答案22。这就是为什么退缩是如此强大。

要使用丢弃模式，我们将对 Listing 5进行以下更改：

// Listing 7. Discarding mode version of early/late firings.
PCollection<KV<String, Integer>> scores = input
  .apply(Window.into(FixedWindows.of(Duration.standardMinutes(2)))
               .triggering(
                 AtWatermark()
                   .withEarlyFirings(AtPeriod(Duration.standardMinutes(1)))
                   .withLateFirings(AtCount(1)))
               .discardingFiredPanes()) // 这一行
  .apply(Sum.integersPerKey());

在具有启发式watermark的流式引擎上再次运行代码将产生如下所示的输出：

图9. 流式引擎上的含有early/later的触发器的丢弃模式版本

虽然输出的整体形状类似于图7的累计模式，但请注意，此丢弃版本中的任何一个窗格都不重叠。因此，每个输出与其他输出是独立的。

如果我们实际看看更正/回撤模式，代码修改类似（但是，请注意，Google Cloud Dataflow的回撤模式仍然处于开发状态，所以这个API中的命名有点推测，不太可能与他相同）：

// Listing 8. Accumulating & retracting mode version of early/late firings.
PCollection<KV<String, Integer>> scores = input
  .apply(Window.into(FixedWindows.of(Duration.standardMinutes(2)))
               .triggering(
                 AtWatermark()
                   .withEarlyFirings(AtPeriod(Duration.standardMinutes(1)))
                   .withLateFirings(AtCount(1)))
               .accumulatingAndRetractingFiredPanes()) // 这一行
  .apply(Sum.integersPerKey());

在流式引擎上运行，结果如下：

[图9. 流式引擎上的含有early/later的触发器的更正模式版本

由于窗口内的窗格相互关联，不太容易看清这些回撤值。回撤值用红色标记了，他与蓝色的窗格重叠，所以看起来像是紫色。每个窗格的两个输出值用逗号分割使他们更容易区分。

让我们把图7（只有启发式的版本），9，10放到一起比较一下：

你可以想到，在存储和计算成本方面，丢弃模式，累计模式，累计&更正模式成本是不断增加的。为此，积累模式的选择也是对正确性，延迟和成本进行权衡。

插曲

至此，我们已经回答了所有四个问题：

What results are calculated?：做什么计算？答案是：通过transform操作。
Where in event time are results calculated?：计算什么时间范围的数据？答案是：通过窗口化。
When in processing time are results materialized? ：何时将计算结果输出？这答案是：使用watermark和trigger配合触发计算。
How do refinements of results relate?：后续数据的处理结果如何影响之前的处理结果？答案是：通过Accumulation模式来决定。

但是我们只看了一种类型的窗口化例子：Event-Time的固定窗口。通过流式101我们知道，还有许多方式进行窗口化，今天我还想讨论其中的两种：Processing-Time的固定窗口和Event-Time的Session窗口。

When/Where: Processing-time windows

Processing-Time窗口十分重要有两个原因：

对于某些场景，例如用量监控（例如，Web服务的流量QPS），您希望在观察到数据流时就分析数据，Processing-Time窗口绝对是合适的方式。
对于事件的实际发生时间很重要的场景（例如，分析用户行为趋势，计费，评分等），使用Processing-Time绝对是错误的选择，能够识别这些场景十分重要！！

因此，值得深入了解Processing-Time窗口和Event-Time窗口之间的不同，特别是考虑到当今大多数流式系统中Processing-Time窗口已经被普及。

当我们面对的模型是严格使用Event-Time作为基础时（例如本文的例子），有两种方式来获得Processing-Time窗口：

触发器：忽略Event-Time（即使用跨所有Event-Time的一个全局窗口，注：也可以理解成没有窗口），并配合使用触发器在Processing-Time轴中定时触发，来提供该窗口的快照。
使用入口时间：当数据到达时，将进入系统的时间赋值给Event-Time，然后使用正常的Event-Time窗口即可。这是Spark Streaming的实现方式。

请注意，这两种方法基本等价，但是在有多个Stage的pipeline的情况下，它们略有不同：在触发器版本，每个Stage都是独立地分割process-time窗口，因此，在第一个Stage的窗口X中出现的数据，可能在下一个Stage的窗口X-1或者窗口X+1中出现。而在使用入口时间作为Processing-Time的版本中，一旦将数据并入到窗口X中，那么在整个pipeline中都会在窗口X中，这是由于不同stage会通过watermark（在DataFlow情况下，Spark Streaming是通过micro-batch的边界来同步）或其他引擎级别的方式来同步处理进度。

使用Processing-Time划分窗口的最大缺点就是：当输入数据被观测到是顺序发生改变时，窗口的内容就改变了。为了用更具体的方式研究这一点，我们比较下面三种场景：

基于Event-Time的窗口，就是我们之前讨论的窗口，不是Processing-Time的，作为比较的基准。
通过触发器实现的Processing-Time窗口
通过入口时间实现的Processing-Time窗口

我们会在这三种场景上分别使用两个不同的数据集（所以，一共会有2*3中情况）。这两个数据集除了观测到的顺序不同，其他均完全相同（比如相同的值发生相同的Event-Time）。第一套数据集就是我们之前一直看到的那个顺序（白色标记）。第二套所有的值都在Processing-Time的轴上移动了（如图12所示，使用紫色标记）。你可以想象一下，只要发生一点意外（比如使用复杂的分布式系统就会打乱顺序），这种情况就会发生。

图12 - 移动输入值在Processing-Time上的位置，同时保持相同的值和Event-Time

Event-time windowing（使用Event-Time窗口化）

为了建立一个比较基线，我们先看一下使用了Event-Time固定窗口搭配启发式watermark的引擎分别作用到这两个数据集上的结果。我们复用Listing5、图7中使用了early/later触发器的代码，得到的结果如下。左侧和我们以前看到的结果一样，右侧是使用了第二种顺序的数据集的结果。重点是：虽然输出过程的形状不一样，但是最终四个窗口的输出结果全部相同：14，22，3和12。

图13.在Event-time固定窗口上使用两个不同Processing-Time顺序的数据集的结果

Processing-time windowing via triggers（使用Trigger实现的Processing-Time窗口化）

好，现在让我们来比较一下使用了Processing-Time的两种方案。首先是Trigger方式。我们从三个角度来看这种方式的Processing-Time窗口是如何工作的：

窗口化方式：由于我们使用Event-Time来模拟Processing-Time，我们使用全局的Event-Time窗口（就是没有窗口）。
触发方式：根据期望的窗口大小，在prcessing-time时间域进行周期性触发。
累加方式：设置窗口的累加方式为『丢弃』，可以让Processing-Time窗口看起来互相独立。（译者注：这里全局的Event-Time窗口会周期性触发多次，但是从Processing-Time角度看一个窗口触发一次，且相互之间没有关系）

相应的代码如Listing9，注意：全局窗口是默认行为，，因此无需指定窗口化策略：

// Listing 9. Processing-time windowing via repeated, discarding panes of a global event-time window.
PCollection<KV<String, Integer>> scores = input
  .apply(Window.triggering(
                  Repeatedly(AtPeriod(Duration.standardMinutes(2))))
               .discardingFiredPanes())
  .apply(Sum.integersPerKey());

当在流式引擎上运行上述代码分别处理上述两组不同观测顺序的数据集，结果如下面图14所示，一些值得注意的点有：

由于我们是通过Event-Time的窗格（Pane）来模拟Processing-Time窗口的，所以图中Processing-Time轴才是窗口，就是图中Y轴的宽度。
由于Processing-Time窗口对输入数据的顺序很敏感，所以，对于观测顺序不同的数据集，窗口的内容不同。左边的结果是12，21，18，右侧的结果是7，26，4。

图14，通过触发器实现的Processing-Time窗口，分别在两组不同观测顺序的数据集上运行

Processing-time windowing via ingress time（使用入口时间实现的Processing-Time窗口化）

最后，让我们看看使用输入数据的入口时间作为Event-Time来实现Processing-Time窗口的方案.在代码方面，这里有四个方面值得一提：

修改Time值：当一个数据到来时，他的Event-Time需要用入口时间来覆盖。注意，DataFlow中还没有对应的统一标准API，以后可能会有（下面的I/O源的代码是虚构的伪代码）。对于 Google Cloud Pub/Sub，您只在发布消息时将timestampLabel的字段设为空即可，其他数据源请参考文档。
窗口化方式：使用标准的Event-Time固定时间窗口即可。
触发方式：由于使用入口时间，所以我们可以使用完美的watermark。因此使用默认触发器即可，它默认可以实现在watermark线通过窗口终点时触发一次。
累加模式：由于每个窗口只有一个输出（译者注：只触发一次，即watermark那一次），所以累加模式不重要，随意设置。

// Listing 10. Explicit default trigger.
PCollection<String> raw = IO.read().withIngressTimeAsTimestamp();
PCollection<KV<String, Integer>> input = raw.apply(ParDo.of(new ParseFn());
PCollection<KV<String, Integer>> scores = input
  .apply(Window.into(FixedWindows.of(Duration.standardMinutes(2))))
  .apply(Sum.integersPerKey());

在流式引擎上的执行过程如下图15所示。当数据到达时，它们的Event-Time被更新以匹配它们的入口时间（即到达时的Processing-Time），随着Processing-Time推移，完美的watermarker线向右水平移位。该图中注意点：

输入数据顺序改变，结果也会改变
与Trigger方式不同，窗口在X轴（Event-Time时间域）中被描绘。但是它表示的含义已经不是Event-Time了。
窗口触发的时间也与Trigger相同，产生的结果也相同，都是：左边12，21，18；右边7，36，4。
由于使用了完美的Watermark，而且X轴表示入口时间，所以，watermark线与理想情况下的watermark线重合，都是向右上方，且斜率为1。

图14，通过使用入口时间实现的Processing-Time窗口，分别在两组不同观测顺序的数据集上运行

虽然使用这两种方式实现Processing-Time窗口很有趣，但是：这里你要明白的最重要的一点是：Event-Time窗口至少在运算结果上是输入数据顺序无关无关（在输入完成之前，实际的Panel会不断变化），而Processing-Time窗口却不是这样。如果你关系事件发生的真实时间，必须使用Event-Time窗口，否则计算结果毫无意义！！。

Where: session windows（会话窗口）

我们就要讲完所有的例子啦。。。如果你读到这里，真的很有耐心。好消息是你的耐心没有白费，下面我会讲一个我最喜欢的特性：数据驱动的动态窗口—Session。请看好了。

Session是一种比较特殊的窗口，它会用某段不活动的间隔为界来捕获一段时间内的事件。在数据分析领域，这种窗口特别有用：它能从用户视角提供用户某段时间内参与了哪些活动。在Session内看到关联的活动，根据会话的长度推断出用户的参与度。

从窗口化的视角，Sesiion在下面这两方面特别有意思：

他们是数据驱动Window的一个例子：窗口的大小和位置都是由输入数据本身决定的，而不像固定/滑动窗口那样有我们预先定义好的时间模式。
他们也是不对齐Window的一个例子：这种窗口没有统一地应用到所有数据上，而只是应用到该数据的一个特定子集（如，每个用户）。把它与对齐Window进行对比：对齐Window会把类似固定/滑动窗口统一地应用到整个数据集上面。

在某些场景下：对于单个Session内的数据，我们可以在时间的前面加一个通用的标志符来标记它属于某个Session（例如：视频播放器的携带服务质量信息心跳包ping。对于某次观看，可以在所有的ping的时间前面加上Session ID）。那么，这个例子的Session十分容易构建了，只要把数据按照Key聚合就是一个Session。

但是，一些更通用的场景（例如：Session自己没法知道它已经开始了）需要这样：Session必须根据数据在时间内的位置来构建。在无序数据集的情况下，这十分棘手。

要提供通用的Session支持，一个核心的洞察是：一个Session Window可以看成是相互重叠的小Window的组合，其中这里的小Window（ proto-session window）比较特别，他们只含有一个Record，窗口大小是inactivity的时间。因此，即使我们看到的数据是无序的，我们可以随着数据一个一个来到，简单地把这些有重叠的小Window合并起来组成一个最终的Session Window。

我们来看一个代码示例，通过使用Listing8的代码（包含了early/late触发器和retract的更新机制）来构建Session：

// Listing 11. Early and late firings with session windows and retractions
PCollection<KV<String, Integer>> scores = input
  .apply(Window.into(Sessions.withGapDuration(Duration.standardMinutes(1))) // 这一行
               .triggering(
                 AtWatermark()
                   .withEarlyFirings(AtPeriod(Duration.standardMinutes(1)))
                   .withLateFirings(AtCount(1)))
               .accumulatingAndRetractingFiredPanes())
  .apply(Sum.integersPerKey());

在流式引擎上允许结果如下图17：

Figure 17 - 具有early/late触发器和retract更新模式的Sessions窗口，一个生动的例子

上图内容是否丰富，我们一起看一下：

当遇到第一个值5的时候，它被放入了一个小Window（起始位置是该Record的Event-Time，大小是inactivity时间，例如1分钟）。以后遇到的任何与这个重叠的Window都会合并到一起。
第二个到的是7，因为它没有与第一个小窗口5重叠，所以类似，它被放入了第二个小Window。
与此同时，watermark线到达了第一个Window终点，所以，第一个窗口5作为预测值在12:06之前被产出。很快第二个watermark也到了，第二个窗口7也作为预测值在12:06之后被产出。
下面陆续看到3，4，3。他们的小窗口相互重叠，最终被合并到了一起。在12:07时遇到了early触发器触发，一个窗口含有值10被输出。
很快8又来了，他与小窗口7和窗口10重叠，他们被合并到一起形成了一个含有值25的Session。当watermark到达这个Session的终点时，值被输出：他具有值25和两个之前的值的retract（7和10）
9到达时也发生了相似的事情，与值5还有值25的Session合并形成了一个更大的值39。在遇到late触发器触发时，39和retract值（5与25）都被输出。

这很强大！更令人惊叹的是：在这么一个模型中描述他们是这么地简单。这个模型把流处理的按纬度分解成可区分可组合的部分的。最后，我们只要关系业务逻辑，而不用操心怎么把数据组合成有用的形式。

如果你不觉得这有多好，看看这篇文章如何使用Spark Streaming手动建立Session 有多痛苦就懂了（注意：我没有职责他们，Spark其实已经做了不错了，现在已经有人努力去写文章说明如何在Spark之上建立一个特定的Session）。当然，他们也没有合适的Event-Time Session，也没有提供early,late的触发机制，还没有retract的更新模式。

终结篇，感觉好极了！

就这些了，讲完了所有的例子，撒花，撒花！你已经有了健壮的流式处理的基础知识，可以出发进入流式的世界。在结束之前，我快速回顾一个我们所涉及的内容，以免你错过了什么。首先，我们所涉及的主要概念：

Event-Time VS Processing-Time（事件时间和处理时间）：最主要的区别是一个是事件发生的真实时间，另一个是数据被数据处理引擎观测到的时间。
Windowing（窗口化）：把数据沿着时间边界分割是处理无界数据的最常用的策略。（无论是Processing-Time还是Event-Time，在DataFlow中我们狭义地定义窗口化是用Event-Time）
Watermarks（水印）：一个时间时间相关的十分有用的概念，提供了在无序/无界数据集上推理数据完整性的一种方式。
Triggers（触发器）：一种告知系统何时可以输出数据的声明机制。在某些场景中十分重要。
Accumulation（累计模式）：当单个窗口需要不断更新多次数据并提供更精确的结果时，不同结果之间如何处理的方式。

其次，我们研究了下面四个问题（我承诺这是最后一次说它们）

What results are calculated?：做什么计算？（计算逻辑是什么，如何表示？） == Transform
Where in event time are results calculated?：计算什么时间范围的数据？ == Windowing
When in processing time are results materialized? ：何时将计算结果输出？ == Watermarks + Triggers
How do refinements of results relate?：后续数据的处理结果如何影响之前的处理结果？ == Accumulation

最终，深入这种流式处理模型的提供的灵活性（因为最终我们就是平衡准确性-Correctness，延迟-Latency和成本-Cost,这些因素关系）。回顾一下：我们只要修改一点点代码就能实现在相同数据集上各种不同的产出：

图18.相同输入数据，对应9种不同的产出

Classic batch:Listing 1 / Figure 2	Fixed windows batch:Listing 2 / Figure 4	Fixed windows streamingwatermark:Listing 2 / Figure 6
Early/late discarding:Listing 7 / Figure 9	Early/late accumulatingListings: 4 & 5 / Figure 7	Early/late retracting:Listing 8 / Figure 10
Processing-time (triggers):Listing 9 / Figure 14	Processing-time: (ingress time)Listing 10 / Figure 15	Sessions:Listing 11 / Figure 17